package cn.xinfei.xdecision.shutdown;

import cn.xinfei.xdecision.conf.ApplicationContextUtil;
import cn.xinfei.xdecision.constant.GlobalConstant;
import cn.xinfei.xdecision.listener.event.CustomerStopEvent;
import cn.xinfei.xdecision.pulsar.consumer.XPulsarConsumerRegister;
import cn.xinfei.xdecision.thread.CustomerThreadPoolExecutor;
import com.alibaba.cloud.nacos.registry.NacosAutoServiceRegistration;
import lombok.extern.slf4j.Slf4j;
import org.apache.catalina.connector.Connector;
import org.slf4j.MDC;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.web.embedded.tomcat.TomcatConnectorCustomizer;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.stereotype.Service;

import java.util.concurrent.*;

@Service
@Slf4j
public class MyShutdown implements ApplicationContextAware, TomcatConnectorCustomizer, ApplicationListener<ContextClosedEvent> {

    private volatile Connector connector;

    private ApplicationContext applicationContext;

    @Autowired
    private NacosAutoServiceRegistration nacosAutoServiceRegistration;


    @Override
    public void customize(Connector connector) {
        this.connector = connector;
    }

    @Override
    public void onApplicationEvent(ContextClosedEvent event) {
        Thread myShutdown = new Thread(() -> {
            MDC.put(GlobalConstant.PID, "MyShutdown");
            MDC.remove(GlobalConstant.NID);
            log.info("==========ContextClosedEvent===MyShutdown================");
            long start = System.currentTimeMillis();
            //nacos
            shutdownNacos();

            //tomcat thread pool
            shutdownTomcatThreadPool();

            //kafka
            shutdownPulsarListener();

            //customer thread pool
            shutdownCustomerThreadPool();

            long end = System.currentTimeMillis();
            MDC.remove(GlobalConstant.NID);
            log.info("-------------------------------------shutdown time={}", end - start);
            MDC.remove(GlobalConstant.PID);
        });
        try {
            myShutdown.setName("myShutdownThread");
            myShutdown.start();
            myShutdown.join(30000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.info("=========================shutdown end============================================");

    }

    public void shutdownNacos() {
        try {
            MDC.put(GlobalConstant.NID, "ShutdownNacos");
            log.info("De-registering-Nacos-start...");
//            NacosAutoServiceRegistration nacosAutoServiceRegistration = ApplicationContextUtil.getApplicationContext().getBean("nacosAutoServiceRegistration", NacosAutoServiceRegistration.class);
            nacosAutoServiceRegistration.destroy();
            try {
                //wait ribbon cache remove
                Thread.sleep(30000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.info("De-registering-Nacos-end...");
        } catch (Exception e) {
            log.info("De-registering-Nacos-end fail", e);
        }
    }


    /**
     * 停止tomcat线程池，
     * 1、先自旋检测线程池里活跃
     * 2、直到没有活跃线程再进行停止
     */
    public void shutdownTomcatThreadPool() {
        try {
            MDC.put(GlobalConstant.NID, "ShutdownTomcatThreadPool");
            log.info("ShutdownTomcatThreadPool start");

            if (this.connector != null) {
                this.connector.pause();
                Executor executor = this.connector.getProtocolHandler().getExecutor();
                if (executor instanceof ThreadPoolExecutor) {
                    ThreadPoolExecutor tomcatThreadPoolExecutor = (ThreadPoolExecutor) executor;
                    tomcatThreadPoolExecutor.shutdown();
                    try {
                        if (!tomcatThreadPoolExecutor.awaitTermination(30, TimeUnit.SECONDS)) {
                            log.warn("Tomcat shutdown shut down gracefully within 30 seconds. Proceeding with forceful shutdown");
                        }
                    } catch (InterruptedException e) {
                        log.error("ShutdownTomcatThreadPool InterruptedException", e);
                    }
                }
            }

            log.info("ShutdownTomcatThreadPool end");
        } catch (Exception e) {
            log.info("ShutdownTomcatThreadPool fail", e);
        }
    }


    public void shutdownCustomerThreadPool() {
        try {
            MDC.put(GlobalConstant.NID, "shutdownCustomerThreadPool");
            log.info("shutdownCustomerThreadPool start");
            CustomerThreadPoolExecutor.customerThreadPoolExecutorMap.entrySet().forEach(e -> {
                String key = e.getKey();
                log.info("shutdownCustomerThreadPool start.threadPool={}",key);
                CustomerThreadPoolExecutor threadPoolExecutor = e.getValue();
//                threadPoolExecutor.shutdown();
                ThreadPoolShutdownHook.shutdownHookByActive(threadPoolExecutor, key);
            });
            log.info("shutdownCustomerThreadPool end");
        } catch (Exception e) {
            log.info("shutdownCustomerThreadPool fail", e);
        }
    }

    public void shutdownPulsarListener() {
        try {
            MDC.put(GlobalConstant.NID, "ShutdownMQ");
            log.info("shutdownPulsarListener start");
            //step1
            applicationContext.publishEvent(new CustomerStopEvent("XStop"));
            try {
                //step2
                XPulsarConsumerRegister XPulsarConsumerRegister = ApplicationContextUtil.getBean("XPulsarConsumerRegister", XPulsarConsumerRegister.class);
                if(null!=XPulsarConsumerRegister) {
                    XPulsarConsumerRegister.stop();
                }
            } catch (Exception e) {
                log.error("shutdownPulsarListener fail", e);
            }

            //step3
            CustomerThreadPoolExecutor pulsarConsumerThreadPool = ApplicationContextUtil.getBean("pulsarConsumerThreadPool", CustomerThreadPoolExecutor.class);
            if (null != pulsarConsumerThreadPool) {
                log.info("shutdownPulsarListener shutdown");
                pulsarConsumerThreadPool.shutdown();
                ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
                executor.schedule(() -> {
                    try {
                        //step4
                        pulsarConsumerThreadPool.shutdownNow();
                        log.info("pulsar delay shutdown success");
                    } catch (Exception e) {
                        log.info("pulsar delay shutdown error.", e);
                    }
                }, 10000, TimeUnit.MILLISECONDS);
            }
            log.info("shutdownPulsarListener end");
        } catch (Exception e) {
            log.info("shutdownPulsarListener fail", e);
        }

    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
